package com.gitee.kamismile.disruptorEx.commandhandling.gateway;

import com.gitee.kamismile.disruptorEx.commandhandling.disruptor.CommandHandlingEntry;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.gitee.kamismile.disruptorEx.commandhandling.CommandCallback;
import com.gitee.kamismile.disruptorEx.commandhandling.CommandMessage;
import com.gitee.kamismile.disruptorEx.commandhandling.callbacks.FutureCallback;
import com.gitee.kamismile.disruptorEx.commandhandling.disruptor.CommandHandlerInvoker;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class DefaultCommandGateway {
    protected final org.slf4j.Logger logger = LoggerFactory.getLogger(this.getClass());
    private Disruptor<CommandHandlingEntry> disruptor;
    private final CommandHandlerInvoker[] commandHandlerInvokers;
    DisruptorConfiguration disruptorConfiguration;

    public DefaultCommandGateway(Disruptor<CommandHandlingEntry> disruptor,DisruptorConfiguration disruptorConfiguration) {


//        Executor executor  = Executors.newCachedThreadPool();
//        disruptor = new Disruptor<CommandHandlingEntry>(CommandHandlingEntry::new, disruptorConfiguration.getBufferSize(), executor,
//                disruptorConfiguration.getProducerType(), disruptorConfiguration.getWaitStrategy());

        this.disruptorConfiguration=disruptorConfiguration;
        this.disruptor=disruptor;
        this.disruptor.setDefaultExceptionHandler(new ExceptionHandler());
        this.commandHandlerInvokers=initializeInvokerThreads();
//        EventPublisher[] publishers =initializePublisherThreads();
//        this.disruptor.handleEventsWith(commandHandlerInvokers).then(publishers);
        this.disruptor.handleEventsWith(commandHandlerInvokers);
        this.disruptor.start();
    }

    public <R> R sendAndWait(CommandMessage command) {
        FutureCallback< Object> futureCallback = new FutureCallback<>();
        send(command, futureCallback);
        return (R) futureCallback.getResult();
    }

    public <R> R sendAndWait(CommandMessage command, long timeout, TimeUnit unit) {
        FutureCallback< Object> futureCallback = new FutureCallback<>();
        send(command, futureCallback);
        return (R) futureCallback.getResult(timeout, unit);
    }

    public  CompletableFuture send(CommandMessage command) {
        FutureCallback<Object> callback = new FutureCallback<Object>();
        send(command, callback);
        return callback;
    }

    public  void sendCallback(CommandMessage command, CommandCallback callback) {
        send(command,callback);
    }

    private void send(CommandMessage command, CommandCallback<Object> futureCallback) {
        RingBuffer<CommandHandlingEntry> ringBuffer = disruptor.getRingBuffer();
        long sequence = ringBuffer.next();
        try {
            CommandHandlingEntry event = ringBuffer.get(sequence);
            event.reset(command,futureCallback);
        } finally {
            ringBuffer.publish(sequence);
        }
    }

//    private EventPublisher[] initializePublisherThreads() {
//        EventPublisher[] publishers = new EventPublisher[0];
//        for (int t = 0; t < publishers.length; t++) {
//            publishers[t] = new EventPublisher();
//        }
//        return publishers;
//    }

    private CommandHandlerInvoker[] initializeInvokerThreads() {
        CommandHandlerInvoker[] invokers;
        invokers = new CommandHandlerInvoker[disruptorConfiguration.getInvokerThreadCount()];
        for (int t = 0; t < invokers.length; t++) {
            invokers[t] = new CommandHandlerInvoker();
        }
        return invokers;
    }

    private class ExceptionHandler implements com.lmax.disruptor.ExceptionHandler {

        @Override
        public void handleEventException(Throwable ex, long sequence, Object event) {
            logger.error("Exception occurred while processing a .", ex);
        }

        @Override
        public void handleOnStartException(Throwable ex) {
            logger.error("Failed to start the DisruptorCommandBus.", ex);
            disruptor.shutdown();
        }

        @Override
        public void handleOnShutdownException(Throwable ex) {
            logger.error("Error while shutting down the DisruptorCommandBus", ex);
        }
    }
}
